package cn.panshi.mongocdc;

import static org.springframework.data.mongodb.core.aggregation.Aggregation.match;
import static org.springframework.data.mongodb.core.aggregation.Aggregation.newAggregation;
import static org.springframework.data.mongodb.core.query.Criteria.where;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.bson.BsonDocument;
import org.bson.BsonValue;
import org.bson.Document;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
import org.springframework.boot.autoconfigure.mongo.MongoAutoConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.core.CollectionOptions;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.aggregation.Aggregation;
import org.springframework.data.mongodb.core.aggregation.AggregationOperation;
import org.springframework.data.mongodb.core.aggregation.AggregationOperationContext;
import org.springframework.data.mongodb.core.aggregation.AggregationOptions;
import org.springframework.data.mongodb.core.aggregation.AggregationResults;
import org.springframework.data.mongodb.core.messaging.ChangeStreamRequest;
import org.springframework.data.mongodb.core.messaging.ChangeStreamRequest.ChangeStreamRequestBuilder;
import org.springframework.data.mongodb.core.messaging.DefaultMessageListenerContainer;
import org.springframework.data.mongodb.core.messaging.MessageListenerContainer;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.ErrorHandler;

import com.alibaba.fastjson.JSONObject;
import com.mongodb.client.MongoCursor;
import com.mongodb.client.model.IndexOptions;
import com.mongodb.client.model.changestream.FullDocument;
import com.mongodb.client.model.changestream.FullDocumentBeforeChange;
import com.mongodb.client.model.changestream.OperationType;

import lombok.extern.slf4j.Slf4j;

@Configuration
@ConditionalOnClass({ IMongoCdcBizRouter.class, IMongoCdcBizRouter.class })
@AutoConfigureAfter({ RedisAutoConfiguration.class, MongoAutoConfiguration.class })
@Slf4j
public class MongoCdcAutoConfiguration {

	@SuppressWarnings("rawtypes")
	@Bean
	//	@ConditionalOnBean(name={ "mongoCdcBizRouter","mongoCdcTtlTaskRouter"}) //这样也是ok的
	@ConditionalOnBean(//
			value = { RedisTemplate.class, IMongoCdcBizRouter.class, IMongoCdcTtlTaskRouter.class, }, //
			name = { "cdcTaskExecutor" })
	@ConditionalOnClass(MongoTemplate.class) //MongoTemplate无法通过@ConditionalOnBean设置条件
	public MongoCdcTtlTaskUtil mongoCdcTtlTaskUtil(//
			RedisTemplate<Object, Object> redisTemplate, //
			MongoTemplate mongoTemplate, //
			IMongoCdcBizRouter mongoCdcBizRouter, //
			IMongoCdcTtlTaskRouter mongoCdcTtlTaskRouter, //
			ThreadPoolTaskExecutor cdcTaskExecutor) {
		log.info("redisTemplate:" + redisTemplate);
		log.info("mongoTemplate:" + mongoTemplate);
		log.info("mongoCdcBizRouter:" + mongoCdcBizRouter);
		log.info("mongoCdcTtlTaskRouter:" + mongoCdcTtlTaskRouter);
		log.info("cdcTaskExecutor:" + cdcTaskExecutor);

		if (!mongoCdcBizRouter.activeByProfile()) {
			log.info("当前profile不启动MongoCdcBizDispatchListener");
			return null;
		}

		MongoCdcBizDispatchListener cdcListner = new MongoCdcBizDispatchListener();
		cdcListner.setMongoCdcBizRouter(mongoCdcBizRouter);
		cdcListner.setMongoCdcTtlTaskRouter(mongoCdcTtlTaskRouter);
		cdcListner.setMongoTemplate(mongoTemplate);
		cdcListner.setRedisTemplate(redisTemplate);

		MessageListenerContainer messageListenerContainer = new DefaultMessageListenerContainer(mongoTemplate,
				cdcTaskExecutor);// Executors.newSingleThreadExecutor()
		messageListenerContainer.start();

		ChangeStreamRequestBuilder<Document> builder = prepareListenerRule(mongoTemplate, cdcListner);

		BsonValue resumeToken = this.queryLatestResumeToken(mongoTemplate);
		if (resumeToken != null) {
			builder.resumeAfter(resumeToken);//
			log.debug("builder.resumeAfter " + resumeToken);
		}
		ChangeStreamRequest<Document> request = builder.build();

		try {
			messageListenerContainer.register(request, Document.class, new ErrorHandler() {
				@Override
				public void handleError(Throwable t) {
					log.error(t.getMessage(), t);
				}
			});
		} catch (Throwable e) {
			log.warn("忽略resumeToken：" + e.getMessage());
			messageListenerContainer.stop();
			messageListenerContainer = new DefaultMessageListenerContainer(mongoTemplate, cdcTaskExecutor);
			messageListenerContainer.start();

			builder = prepareListenerRule(mongoTemplate, cdcListner);

			messageListenerContainer.register(request, Document.class);
		}

		this.createTtlIndexIfNotExist(mongoTemplate);

		this.createCdcOptLogTableIfNotExist(mongoTemplate);

		return new MongoCdcTtlTaskUtil(mongoTemplate);
	}

	private ChangeStreamRequestBuilder<Document> prepareListenerRule(MongoTemplate mongoTemplate,
			MongoCdcBizDispatchListener cdcListner) {
		ChangeStreamRequestBuilder<Document> builder = ChangeStreamRequest.builder(cdcListner)
				.database(mongoTemplate.getDb().getName())
				//				.collection("XxxColl") // 需要监听的集合名，不指定默认监听数据库的
				.filter(newAggregation(match(where("operationType").in(//
						OperationType.INSERT.getValue(), OperationType.UPDATE.getValue(),
						OperationType.REPLACE.getValue(), OperationType.DELETE.getValue()))))
				/**
				 * MongoDB6.0支持preImage  
				 * https://www.alexbevi.com/blog/2020/07/15/mongodb-4-dot-4-change-streams-and-pre-image-support/
				 * https://www.mongodb.com/docs/v6.0/changeStreams/#change-streams-with-document-pre--and-post-images
				 */
				.fullDocumentBeforeChangeLookup(FullDocumentBeforeChange.WHEN_AVAILABLE)

				.fullDocumentLookup(FullDocument.UPDATE_LOOKUP); // 不设置时，文档更新时，只会发送变更字段的信息，设置UPDATE_LOOKUP会返回文档的全部信息

		return builder;
	}

	private void createCdcOptLogTableIfNotExist(MongoTemplate mongoTemplate) {
		if (!mongoTemplate.collectionExists(CdcOptLogDoc.TABLE)) {
			mongoTemplate.createCollection(CdcOptLogDoc.TABLE,
					CollectionOptions.empty().capped().maxDocuments(10000).size(10240000));
		}

		MongoCursor<Document> it = mongoTemplate.getCollection(CdcOptLogDoc.TABLE).listIndexes().iterator();
		boolean lastUpdateIndexExist = false;
		while (it.hasNext()) {
			Document index = it.next();
			Document indexKey = index.get("key", Document.class);
			if (indexKey.get("lastUpdate") != null) {
				log.debug("存在lastUpdate字段相关索引");
				lastUpdateIndexExist = true;
			}
		}

		if (!lastUpdateIndexExist) {
			mongoTemplate.getCollection(CdcOptLogDoc.TABLE).createIndex(new Document("lastUpdate", 1));
			log.info("创建 CdcOptLogDoc lastUpdate index");
		}

		/*Document executeCommand = mongoTemplate.executeCommand(
				new Document("convertToCapped", CdcOptLogDoc.TABLE).append("size", 10240000).append("max", 10000));
		
		log.info(executeCommand + "");*/

		//需要高权限
		//		executeCommand = mongoTemplate.executeCommand(new Document("collMod", CdcOptLogDoc.TABLE).append("max", 200));
		//		log.info(executeCommand + "");
	}

	private void createTtlIndexIfNotExist(MongoTemplate mongoTemplate) {
		boolean ttlIndexExist = false;
		boolean triggerIdIndexExist = false;

		MongoCursor<Document> it = mongoTemplate.getCollection(CdcTtlTaskDoc.TABLE).listIndexes().iterator();
		while (it.hasNext()) {
			Document index = it.next();
			Document indexKey = index.get("key", Document.class);
			if (indexKey.get("ttlTime") != null) {
				log.debug("存在ttlTime字段相关索引");
				ttlIndexExist = true;
			} else if (indexKey.get("triggerId") != null && indexKey.size() == 1) {//
				log.debug("存在triggerId字段的非复合索引");
				triggerIdIndexExist = true;
			}
		}

		if (ttlIndexExist) {
			log.info("ttlTime index已经存在");
		} else {
			log.info("ttlTime index不存在");
			Document document = new Document();
			document.append("ttlTime", 1);
			IndexOptions indexOptions = new IndexOptions().expireAfter(0L, TimeUnit.SECONDS);
			mongoTemplate.getCollection(CdcTtlTaskDoc.TABLE).createIndex(document, indexOptions);
			log.info("创建 ttlTime index expireAfter 0s：ttl索引 超时时间取决于对应记录字段的date值");
		}
		if (triggerIdIndexExist) {
			log.info("triggerId index已经存在");
		} else {
			Document document = new Document();
			document.append("triggerId", 1);
			IndexOptions indexOptions = new IndexOptions().unique(true).sparse(true);
			mongoTemplate.getCollection(CdcTtlTaskDoc.TABLE).createIndex(document, indexOptions);
			log.info("创建 triggerId index unique sparse：唯一索引 同时允许为空");
		}
	}

	private BsonValue queryLatestResumeToken(MongoTemplate mongoTemplate) {
		List<Document> aggOperationList = Arrays.asList(
				new Document("$match", new Document("lastUpdate", new Document("$exists", 1L))),
				new Document("$sort", new Document("lastUpdate", -1L)), new Document("$limit", 1L),
				new Document("$project", new Document("_data", "$resumeToken").append("_id", 0L)));

		AggregationResults<Document> aggregateQuery = aggregateQuery(mongoTemplate, CdcOptLogDoc.TABLE,
				aggOperationList);

		Iterator<Document> it = aggregateQuery.iterator();
		if (it.hasNext()) {
			return BsonDocument.parse(it.next().toJson());
		} else {
			return null;
		}
	}

	public static AggregationResults<Document> aggregateQuery(MongoTemplate mongoTemplate, String collection,
			List<Document> aggOperationList) {

		if (log.isTraceEnabled()) {
			String json = JSONObject.toJSONString(aggOperationList, true);
			log.trace(json);
		}
		//转换成Aggregation
		List<CustomGroupOperation> custAggOperationList = new ArrayList<>();
		for (Document doc : aggOperationList) {
			custAggOperationList.add(new CustomGroupOperation(doc));
		}
		Aggregation aggList = Aggregation.newAggregation(custAggOperationList)
				.withOptions(AggregationOptions.builder().allowDiskUse(true).build());

		//聚合查询
		return mongoTemplate.aggregate(aggList, collection, Document.class);
	}

	static class CustomGroupOperation implements AggregationOperation {
		private Document operation;

		public CustomGroupOperation(Document operation) {
			this.operation = operation;
		}

		@Override
		public Document toDocument(AggregationOperationContext context) {
			return context.getMappedObject(operation);
		}
	}
}